use std::io::Write;
use std::net::TcpStream;
// use std::thread;
use std::time::Duration;

use anyhow::Result;
use async_channel::Sender;
use blocking::unblock;
use futures::future::Either;
use futures::{future, io, AsyncReadExt, AsyncWriteExt, Future, FutureExt};
use smol::Task;
use smol::{Async, Timer};

fn main() {
    // // Same number of threads as there are CPU cores.
    // let num_threads = num_cpus::get().max(1)-1;
    // println!("num_threads = {:?}", num_threads);
    //
    // // Run the thread-local and work-stealing executor on a thread pool.
    // for _ in 0..num_threads {
    //     // A pending future is one that simply yields forever.
    //     thread::spawn(|| smol::run(future::pending::<()>()));
    // }

    // Set a handler that sends a message through a channel.
    let (s, ctrl_c) = async_channel::bounded::<()>(1);
    let quit = s.clone();
    let handle = move || {
        let _ = quit.send(()).now_or_never();
    };
    ctrlc::set_handler(handle).unwrap();

    let task = Task::spawn(run(s));

    smol::run(async {
        if let Err(e) = ctrl_c.recv().await {
            println!("wait ctrlc fail: {:?}", e);
        }
        task.cancel().await;
        println!("Program quit.");
    });
}

async fn run(quit: Sender<()>) {
    let mut addr = String::new();
    let mut seq = 1_usize;

    // loop until quit
    loop {
        if addr.is_empty() {
            match get_server().await {
                Ok(s) => addr = s,
                Err(e) => {
                    println!("Error: {:?}", e);
                    break;
                }
            }
            continue;
        }

        let addr = addr.clone();
        if let Err(e) = handle(addr, seq).await {
            println!("{:?}", e);
            break;
        }
        seq += 1;
    }
    let _ = quit.send(()).now_or_never();
}

async fn handle(addr: String, seq: usize) -> Result<()> {
    let req = get_msg().await?;
    Task::spawn(async move {
        if let Err(e) = send(seq, addr, req).await {
            println!("第{}次请求 {:?}", seq, e);
        }
    })
    .detach();
    Ok(())
}

async fn get_server() -> Result<String> {
    let buf = unblock(|| -> Result<String> {
        std::io::stdout().write_all(
            "请输入调用的服务器：1-ESB（10.1.77.21:9025） 2-Gateway180（10.1.68.180:9000）\n"
                .as_bytes(),
        )?;
        std::io::stdout().flush()?;

        let mut buf = String::new();
        std::io::stdin().read_line(&mut buf)?;
        Ok(buf)
    })
    .await?;

    match buf.lines().next().unwrap_or("") {
        "0" => Ok("127.0.0.1:9000".into()),
        "1" => Ok("10.1.77.21:9025".into()),
        "2" => Ok("10.1.68.180:9000".into()),
        q if q.eq("quit") => anyhow::bail!("user quit."),
        o => Ok(o.into()),
    }
}

async fn get_msg() -> Result<String> {
    let mut msg = String::new();
    loop {
        let buf = unblock(|| -> Result<String> {
            std::io::stdout().write_all("请输入请求报文\n".as_bytes())?;
            std::io::stdout().flush()?;

            let mut buf = String::new();
            std::io::stdin().read_line(&mut buf)?;
            buf = buf.lines().next().unwrap_or("").into();
            Ok(buf)
        })
        .await?;

        if buf.is_empty() {
            break;
        } else if buf.eq("quit") {
            anyhow::bail!("user quit")
        }

        msg.push_str(&*buf);
    }
    println!("报文读取完毕，准备发送");

    Ok(msg)
}

async fn send(seq: usize, addr: String, req: String) -> Result<String> {
    let mut res = format!("{:010}{}", req.len(), req);
    let time = Duration::from_secs(30);
    let mut stdout = smol::writer(std::io::stdout());

    // println!("第{}次请求，连接服务器：{}\n", seq, addr);
    stdout
        .write_all(format!("第{}次请求，连接服务器：[{}]\n", seq, addr).as_bytes())
        .await?;
    stdout.flush().await?;
    let mut stream = timeout(time, Async::<TcpStream>::connect(addr)).await?;
    stream.write_all(res.as_bytes()).await?;
    // println!("第{}次请求，发送数据：{}\n", seq, res);
    stdout
        .write_all(format!("第{}次请求，发送数据：[{}]\n", seq, res).as_bytes())
        .await?;
    stdout.flush().await?;
    res.clear();

    timeout(time, stream.read_to_string(&mut res)).await?;
    stream.close().await?;
    // println!("第{}次请求，接收响应：{}\n", seq, res);
    stdout
        .write_all(format!("第{}次请求，接收响应：[{}]\n", seq, res).as_bytes())
        .await?;
    stdout.flush().await?;
    Ok(res)
}

async fn timeout<T>(dur: Duration, f: impl Future<Output = io::Result<T>>) -> io::Result<T> {
    futures::pin_mut!(f);
    match future::select(f, Timer::after(dur)).await {
        Either::Left((out, _)) => out,
        Either::Right(_) => Err(io::ErrorKind::TimedOut.into()),
    }
}
